package com.study.base.io.nio.mainsubreactor;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

/**
 * Created by tan on 2017/7/18.
 */
public class Acceptor implements Runnable {
	private ServerSocketChannel serverSocket; // mainReactor socket通道
	private final int cores = Runtime.getRuntime().availableProcessors(); // 取得CPU核心數
	private final Selector[] selectors = new Selector[cores]; // 创建核心数个selector給subReactor用

	private SubReactor[] subReactors = new SubReactor[cores]; // subReactor
	private int next = 0; // 当前可使用的subReactor索引
	private Thread[] threads = new Thread[cores]; // subReactor线程

	public Acceptor(ServerSocketChannel serverSocket) throws IOException {
		this.serverSocket = serverSocket;

		// 创建selector以及多个subReactor线程
		for (int i = 0; i < cores; i++) {
			selectors[i] = Selector.open();
			subReactors[i] = new SubReactor(selectors[i], serverSocket, i);
			threads[i] = new Thread(subReactors[i]);
			threads[i].start();
		}
	}

	@Override
	public synchronized void run() {//注意加synchronized
		try {
			SocketChannel socket = serverSocket.accept();
			if (socket != null) {
				socket.configureBlocking(false);
				subReactors[next].setRestart(true);//暂停线程
				selectors[next].wakeup();//使一个阻塞住的selector.select()操作立即返回

				SelectionKey key = serverSocket.register(selectors[next], SelectionKey.OP_READ);

				selectors[next].wakeup();//使一个阻塞住的selector.select()操作立即返回
				subReactors[next].setRestart(false);//启动线程

				key.attach(new Handler(selectors[next], socket));
				if (++next == selectors.length)
					next = 0;
			}
		} catch (IOException e) {
			e.printStackTrace();
		}

	}
}
